package com.aibaixun.web.controller.listen;

import java.time.LocalDateTime;
import java.time.ZoneId;
import java.time.ZonedDateTime;
import java.time.format.DateTimeFormatter;
import java.util.Map;
import java.util.Optional;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import cn.hutool.json.JSONUtil;
import com.aibaixun.common.utils.StringUtils;
import com.aibaixun.system.domain.ActiveAlarm;
import com.aibaixun.system.domain.AlarmSourceEntity;
import com.aibaixun.system.service.DicmAlarmService;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import lombok.NonNull;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringEscapeUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.TopicPartition;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.listener.ConsumerSeekAware;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;

/**
 * kafka消息消费
 * @author zhangyh
 */
@Slf4j
@Component
public class DicmKafkaConsumer implements ConsumerSeekAware{
    @Autowired
    private DicmAlarmService dicmAlarmService;

    private static DicmKafkaConsumer dicmKafkaConsumer;

    @PostConstruct
    public void init(){
        dicmKafkaConsumer = this;
        dicmKafkaConsumer.dicmAlarmService = this.dicmAlarmService;
    }

    @Override
    public void onPartitionsAssigned(Map<TopicPartition, Long> assignments,
                                     @NonNull ConsumerSeekAware.ConsumerSeekCallback callback) {
        assignments.keySet().forEach(topicPartition-> callback.seekToEnd(topicPartition.topic(),
            topicPartition.partition()));
    }

    @KafkaListener(
        topics = {"dcim_alarm","gz-alarm","dh-alarm"},
        containerFactory = "containerFactory"
    )
    public void listen(ConsumerRecord<?, ?> record) {
        System.out.println("收到kafka推送信息");
//        Optional<?> kafkaMessage = Optional.ofNullable(record.value());
        Object messageData = record.value();
        if (messageData != null) {
            //kafka消息处理
            try {
                String message = messageData.toString();
                if(StringUtils.isBlank(message)){
                    return;
                }
                if(message.startsWith("\"") && message.endsWith("\"")){
                    message = message.substring(1,message.length()-1);
                }
                String topic = record.topic();
                String regex = "\\\\";
                message = message.replaceAll(regex, "");
                if(message.contains("\\")){
                    message = message.replaceAll(regex, "");
                }
                JSONObject jsonObject = new JSONObject();
//                if(StringUtils.equals("gz-alarm",topic)){
//                    AlarmSourceEntity alarmSourceEntity = JSON.parseObject(message,AlarmSourceEntity.class);
//                    jsonObject = JSONObject.parseObject(JSONObject.toJSONString(alarmSourceEntity));
//                }else if(StringUtils.equals("dh-alarm",topic)){
//                    ActiveAlarm activeAlarm = JSON.parseObject(message,ActiveAlarm.class);
//                    jsonObject = JSONObject.parseObject(JSONObject.toJSONString(activeAlarm));
//                }else{
//                    jsonObject = JSONObject.parseObject(message);
//                }
                jsonObject = JSONObject.parseObject(message);
                jsonObject.put("kafkaTopic",topic);
                System.out.println("保存告警数据参数："+jsonObject);
                dicmAlarmService.handleAlarmMsg(jsonObject);
            }catch (Exception e){
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) {
        String s3 = "2023-10-22 21.38.21";
        System.out.println(LocalDateTime.parse(s3,DateTimeFormatter.ofPattern("yyyy-MM-dd HH.mm.ss")));

        String s = "{'SerialNo'.236840,'SiteID'.'3087007744','SynNo'.41137444,'DeviceID'.'3908435968','AlarmDesc'.'单体31电压过低告警.11.855000','AlarmValue'.11.855,'AlarmLevel'.2,'NMAlarmID'.'609-007-00-007005','SCID'.'1','SignalID'.'007005','Type'.2,'AlarmTime'.'2023-10-22 20.02.41','SignalName'.'单体31电压过低告警','AlarmStatus'.0,'Id'.41137444,'SignalNumber'.'031'}";
        s = s.replace("\"", "");
        s = s.replace("'", "");
        s = s.replace("{", "");
        s = s.replace("}", "");
        String[] sList = s.split(",");
        JSONObject jsonObject = new JSONObject();
        for (String s1 : sList) {
            int index = s1.indexOf(".",0);
            jsonObject.put(s1.substring(0,index),s1.substring(index+1));
        }
        s = s.replace(".", ":");
    }

}


